package cn.gwm.flink.streaming.sink.kafka;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.utils.ConfigLoader;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

/**
 * @Author: liangguang
 * @Date: 2022/10/12
 * @Description: 自定义 flink kafka 相关工具类
 */
public class FlinkKafkaUtil {


    public static FlinkKafkaConsumer<String> getConsumerGroupOffsets(String groupId, String topic) {
        FlinkKafkaConsumer<String> kafkaConsumer = getConsumer(ConfigLoader.get("bootstrap.servers"), groupId, topic);
        kafkaConsumer.setStartFromGroupOffsets();
        return kafkaConsumer;
    }

    public static FlinkKafkaConsumer<String> getConsumerEarLiest(String groupId, String topic) {
        FlinkKafkaConsumer<String> kafkaConsumer = getConsumer(ConfigLoader.get("bootstrap.servers"), groupId, topic);
        kafkaConsumer.setStartFromEarliest();
        return kafkaConsumer;
    }

    public static FlinkKafkaConsumer<String> getConsumerLatest(String groupId, String topic) {
        FlinkKafkaConsumer<String> kafkaConsumer = getConsumer(ConfigLoader.get("bootstrap.servers"), groupId, topic);
        kafkaConsumer.setStartFromLatest();
        return kafkaConsumer;
    }

    /**
     * 获取 FlinkKafkaConsumer  基础消费者信息，可自定义配置
     *
     * @param servers
     * @param groupId
     * @param topic
     * @return
     */
    public static FlinkKafkaConsumer<String> getConsumer(String servers, String groupId, String topic) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        // 默认  500
        props.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "200");
        final FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), props);
        return kafkaConsumer;
    }

    /**
     * tokafka 生产者
     *
     * @param servers         kafka地址
     * @param topic
     * @param transactionalId
     * @return
     */
    public static FlinkKafkaProducer toKafka(String servers, String topic, Properties properties, String transactionalId) {
        Properties prodProps = new Properties();
        prodProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        // acks  0,1,all
        prodProps.setProperty(ProducerConfig.ACKS_CONFIG, "all");
//        //发送频率
//        prodProps.setProperty(ProducerConfig.LINGER_MS_CONFIG, "300");
//        //批量发送
//        prodProps.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "1638400");
        // 事务超时时间
        prodProps.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000L + "");
        if (transactionalId != null) {
            prodProps.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tranId-" + transactionalId + "-" + topic);
        }
        prodProps.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        prodProps.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1");

        if (properties != null) {
            for (String key : properties.stringPropertyNames()) {
                prodProps.setProperty(key, properties.getProperty(key));
            }
        }

        FlinkKafkaProducer myProducer = KafkaSinkProducer.producerBuilder(topic, prodProps, BaseFields.vin);
        return myProducer;
    }


    /**
     * 默认 推送 服务器 bootstrap.servers 配置
     *
     * @param topic
     * @return
     */
    public static FlinkKafkaProducer toKafka(String topic) {
        return toKafka(ConfigLoader.get("bootstrap.servers"), topic, null, null);
    }

    public static FlinkKafkaProducer toKafka(String services, String topic) {
        return toKafka(services, topic, null, null);
    }

    /**
     * @param topic           主题
     * @param transactionalId 事务id
     * @return
     */
//    public static FlinkKafkaProducer toKafka(String topic, String transactionalId) {
//        return toKafka(ConfigLoader.get("bootstrap.servers"), topic, null, transactionalId);
//    }

    /**
     * @param topic           主题
     * @param transactionalId 事务id
     * @return
     */
    public static FlinkKafkaProducer toKafka(String topic, Properties properties, String transactionalId) {
        return toKafka(ConfigLoader.get("bootstrap.servers"), topic, properties, transactionalId);
    }
}
